package com.chukun.flink.stream.action.fraud.rules.major;

import com.chukun.flink.stream.action.fraud.config.Config;
import com.chukun.flink.stream.action.fraud.config.Parameters;
import com.chukun.flink.stream.action.fraud.rules.functions.AverageAggregate;
import com.chukun.flink.stream.action.fraud.rules.functions.DynamicAlertFunction;
import com.chukun.flink.stream.action.fraud.rules.functions.DynamicKeyFunction;
import com.chukun.flink.stream.action.fraud.rules.functions.SimpleBoundedOutOfOrdernessTimestampExtractor;
import com.chukun.flink.stream.action.fraud.rules.model.Alert;
import com.chukun.flink.stream.action.fraud.rules.model.Keyed;
import com.chukun.flink.stream.action.fraud.rules.model.Rule;
import com.chukun.flink.stream.action.fraud.rules.model.Transaction;
import com.chukun.flink.stream.action.fraud.rules.sink.AlertsSink;
import com.chukun.flink.stream.action.fraud.rules.sink.CurrentRulesSink;
import com.chukun.flink.stream.action.fraud.rules.sink.LatencySink;
import com.chukun.flink.stream.action.fraud.rules.source.RulesSource;
import com.chukun.flink.stream.action.fraud.rules.source.TransactionsSource;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.io.IOException;
import java.util.concurrent.TimeUnit;

/**
 * @author chukun
 * @version 1.0.0
 * @description 规则处理器
 * @createTime 2022年05月22日 11:20:00
 */
public class RulesEvaluator {

    private Config config;

    public RulesEvaluator(Config config) {
        this.config = config;
    }

    public void run() throws Exception {
        // 创建flink运行时环境
        StreamExecutionEnvironment env = configureStreamExecutionEnvironment();

        DataStream<Rule> rulesUpdateStream = getRulesUpdateStream(env);

        DataStream<Transaction> transactionsStream = getTransactionsStream(env);

        BroadcastStream<Rule> rulesStream = rulesUpdateStream.broadcast(Descriptors.rulesDescriptor);

        SingleOutputStreamOperator<Alert> alertStream = transactionsStream.connect(rulesStream)
                .process(new DynamicKeyFunction())
                .uid("DynamicKeyFunction")
                .name("Dynamic Partitioning Function")
                .keyBy(Keyed::getKey)
                .connect(rulesStream)
                .process(new DynamicAlertFunction())
                .uid("DynamicAlertFunction")
                .name("Dynamic Rule Evaluation Function");

        // 获取延迟的数据流
        DataStream<Long> latency =
                alertStream.getSideOutput(Descriptors.latencySinkTag);

        // 获取正常的数据流
        DataStream<Rule> currentRules =
                alertStream.getSideOutput(Descriptors.currentRulesSinkTag);

        DataStream<String> alertsJson = AlertsSink.alertsStreamToJson(alertStream);
        DataStream<String> currentRulesJson = CurrentRulesSink.rulesStreamToJson(currentRules);

        int sinkParallelism = config.get(Parameters.SINK_PARALLELISM);

        alertsJson
                .addSink(AlertsSink.createAlertsSink(config))
                .setParallelism(sinkParallelism)
                .name("Alerts JSON Sink");
        currentRulesJson
                .addSink(CurrentRulesSink.createRulesSink(config))
                .setParallelism(sinkParallelism)
                .name("Rules Export Sink");

        DataStream<String> latencies =
                latency
                        .windowAll(SlidingProcessingTimeWindows.of(Time.seconds(10), Time.seconds(0)))
                        .aggregate(new AverageAggregate())
                        .map(String::valueOf);
        latencies.addSink(LatencySink.createLatencySink(config)).name("Latency Sink");

        env.execute("Fraud Detection Engine");


    }

    /**
     * 创建flink运行环境
     * @return
     */
    private StreamExecutionEnvironment configureStreamExecutionEnvironment() {
        final String localMode = config.get(Parameters.LOCAL_EXECUTION);

        StreamExecutionEnvironment env;

        if (localMode.isEmpty() || localMode.equals(Parameters.LOCAL_MODE_DISABLE_WEB_UI)) {
            env = StreamExecutionEnvironment.getExecutionEnvironment();
        } else {
            Configuration finkConfig = new Configuration();
            finkConfig.set(RestOptions.BIND_PORT, localMode);
            env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(finkConfig);
        }
        if (!localMode.isEmpty()) {
            // slower restarts inside the IDE and other local runs
            env.setRestartStrategy(
                    RestartStrategies.fixedDelayRestart(
                            10, org.apache.flink.api.common.time.Time.of(10, TimeUnit.SECONDS)));
        }

        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.getCheckpointConfig().setCheckpointInterval(config.get(Parameters.CHECKPOINT_INTERVAL));
        env.getCheckpointConfig()
                .setMinPauseBetweenCheckpoints(config.get(Parameters.MIN_PAUSE_BETWEEN_CHECKPOINTS));
        return env;
    }

    /**
     * 获取交易的数据流
     * @param env
     * @return
     */
    private DataStream<Transaction> getTransactionsStream(StreamExecutionEnvironment env) {
        // Data stream setup
        SourceFunction<String> transactionSource = TransactionsSource.createTransactionsSource(config);
        int sourceParallelism = config.get(Parameters.SOURCE_PARALLELISM);
        DataStream<String> transactionsStringsStream =
                env.addSource(transactionSource)
                        .name("Transactions Source")
                        .setParallelism(sourceParallelism);
        DataStream<Transaction> transactionsStream =
                TransactionsSource.stringsStreamToTransactions(transactionsStringsStream);
        return transactionsStream.assignTimestampsAndWatermarks(
                new SimpleBoundedOutOfOrdernessTimestampExtractor<>(config.get(Parameters.OUT_OF_ORDERNESS)));
    }

    /**
     * 创建规则流
     * @param env
     * @return
     * @throws IOException
     */
    private DataStream<Rule> getRulesUpdateStream(StreamExecutionEnvironment env) throws IOException {
        RulesSource.Type rulesSourceType = getRulesSourceType();
        SourceFunction<String> rulesSource = RulesSource.createRulesSource(config);
        DataStream<String> ruleStrStream = env.addSource(rulesSource).name(rulesSourceType.getName()).setParallelism(1);
        return RulesSource.stringsStreamToRules(ruleStrStream);
    }

    /**
     * 获取规则类型
     * @return
     */
    private RulesSource.Type getRulesSourceType() {
        String rulesSource = config.get(Parameters.RULES_SOURCE);
        return RulesSource.Type.valueOf(rulesSource.toUpperCase());
    }
}
